﻿package org.aaa.cloudagent;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;  
import java.util.ArrayList;  
import java.util.List;

import org.apache.commons.cli.ParseException;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.XMLConfiguration;
import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;  
import org.apache.hadoop.hbase.HColumnDescriptor;  
import org.apache.hadoop.hbase.HTableDescriptor;  
import org.apache.hadoop.hbase.KeyValue;  
import org.apache.hadoop.hbase.MasterNotRunningException;  
import org.apache.hadoop.hbase.ZooKeeperConnectionException;  
import org.apache.hadoop.hbase.client.Delete;  
import org.apache.hadoop.hbase.client.Get;  
import org.apache.hadoop.hbase.client.HBaseAdmin;  
import org.apache.hadoop.hbase.client.HTable;  
import org.apache.hadoop.hbase.client.HTablePool;  
import org.apache.hadoop.hbase.client.Put;  
import org.apache.hadoop.hbase.client.Result;  
import org.apache.hadoop.hbase.client.ResultScanner;  
import org.apache.hadoop.hbase.client.Scan;  
import org.apache.hadoop.hbase.filter.Filter;  
import org.apache.hadoop.hbase.filter.FilterList;  
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;  
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;  
import org.apache.hadoop.hbase.util.Bytes;  
  
public class DocOperater {  
  
    public static Configuration configuration;  
    private static boolean bSetConfigurationFlag = false;
    public static HBaseManager hbasemanager=null;
    
    public static String InfoTableName = "t_docinfo";
    public static String IndexTableName = "t_docinfo_idx"; 
    public static String InfoTable_ColName_docinfo = "docinfo";
    public static String InfoTable_ColName_docsave = "docsave"; 
    public static String IndexTable_ColName_docidset = "docidset";
    private static XMLConfiguration presetDataCfg = null;
    private static String PresetdataBaseKey = null;
    private static ArrayList<TableSchema> presetdataList = null;
    public static String FuseMntPoint = null;
    static{
    	hbasemanager= new HBaseManager();
    }
    
    public DocOperater()
    {

    }
  
    
    
    /*
     * 对外接口, 根据shema_filename的xml文件创建表
     */
    public static Configuration GetConfigurationFromSchema(String schema_filename)
    {
    	try
    	{
    		System.out.println("schema:"+schema_filename);
    		hbasemanager.readConfiguration(schema_filename);
            
    		configuration = hbasemanager.getConfiguration();
    		    		
    		return configuration;
       	}
    	catch (final Exception e)
    	{
            e.printStackTrace();
    	}
		return configuration;
    }
    
    /*
     * 对外接口, 根据shema_filename的xml文件创建表
     */
    public static Configuration LoadSchema(String schema_filename)
    {
    	try
    	{
    		System.out.println("schema:"+schema_filename);
    		hbasemanager.readConfiguration(schema_filename);
            
    		configuration = hbasemanager.getConfiguration();
    		FuseMntPoint = hbasemanager.getFuseMntPointer();
    		return configuration;
       	}
    	catch (final Exception e)
    	{
            e.printStackTrace();
    	}
		return configuration;
    }
    
    
    /*
     * 对外接口, 根据shema_filename的xml文件创建表
     */
    public static void RecreateOrUpdateTableBySchema(boolean bRecreate)
    {
    	try
    	{
    		hbasemanager.RecreateOrUpdateTableBySchema(bRecreate);
            System.out.println("done.");
       	}
    	catch (final Exception e)
    	{
            e.printStackTrace();
    	}
    }
    

    
    

  /*
    public static void main(String[] args) { 
    	

    	System.out.println("CreateTableBySchema");
    	
    	LoadSchema("dbschema.xml");
    	CreateTableBySchema();
    	
        try {
			LoadPresetVideoData("presetdata.xml");
		} catch (ConfigurationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		__testQuery(); 
	}
    */	
     
    public static void __testQuery() 
    { 
        
        /*
         System.out.println("\n-------------------获得列表并只获取第一个-------------------------------------");
    	String keywordtest = "新业务5";
    	String[] docidlist = GetDocidlistByKeyword(keywordtest);
    	
    	if(null != docidlist[0])
    	{
			String docpath = GetDirByDocid(docidlist[0]);
    		System.out.println("Get docpath "+docpath+" by :"+docidlist[0] + " by "+ keywordtest);
    	}
         
                  
         System.out.println("\n-------------------根据关键词获得所有详细信息-------------------------------------");
         StringBuffer sResult = new StringBuffer("");
         GetDocinfoListByKeyword("actor", sResult);
         
         System.out.println("main Get result:"+sResult);
         
         
         System.out.println("\n-------------------根据关键词获得所有详细信息-------------------------------------");
         StringBuffer sResult2 = new StringBuffer("");
         GetDocinfoListByKeyword("协议", sResult);
         
         System.out.println("main Get result:"+sResult);
         */
        
    }  

    
  //对外接口   根据关键词查询包含此关键词的docid, 返回的是String数组
    public static String[] GetDocidlistByKeyword(String keyword)
    {
        String sDocid = null;
        
        HTablePool pool = new HTablePool(configuration, 1000);  
        HTable table = (HTable) pool.getTable(IndexTableName);  
        try {  
            Get scan = new Get(keyword.getBytes());// 根据rowkey查询
            scan.addColumn(IndexTable_ColName_docidset.getBytes(), "docidset".getBytes());
            Result r = table.get(scan);  
            if(r.isEmpty())
            {
            	System.out.println("return NULL get by " + keyword +" from "+IndexTableName);
            	return null;
            }
            else
            {
	            System.out.println();  
	            sDocid = new String(r.list().get(0).getValue()); 
	            System.out.println("query "+IndexTableName+" by:" + keyword +" return :" + sDocid);            
          	
            }

        } catch (IOException e) {  
            e.printStackTrace();  
        }
        
        return sDocid.split(";");
   	
    }
    
    //对外接口  根据一个docid返回一个路径
    public static String GetDirByDocid(String docid)
    {
    	String docpath="";
        HTablePool pool = new HTablePool(configuration, 1000);  
        HTable table = (HTable) pool.getTable(InfoTableName);  
        try {  
            Get scan = new Get(Bytes.toBytes(docid));// 根据rowkey查询  
            scan.addColumn("docsave".getBytes(), "docpath".getBytes()); 
            Result r = table.get(scan);  
            if(r.isEmpty())
            {
            	System.out.println("return NULL get by " + docid +" from "+InfoTableName);
            	return "";
            }
            else
            {
	            System.out.println("GetDirByDocid docid=" + new String(r.getRow()));  
	            docpath = new String(r.list().get(0).getValue());
	            System.out.println("return docpath:" + docpath);            
          	
            }

        } catch (IOException e) {  
            e.printStackTrace();  
        }
        
        System.out.println("return docpath:" + FuseMntPoint+docpath);  
		return FuseMntPoint+docpath+"/"; 
    	
    }
    
    //对外接口  不好, 暂时保留
    public static String __GetDirlistByDocidlist(String docids)
    {
    	String docpath="";
        HTablePool pool = new HTablePool(configuration, 1000);  
        HTable table = (HTable) pool.getTable(InfoTableName);  
        
        // To be Modify
        String docid = docids;
        try {  
            Get scan = new Get(docid.getBytes());// 根据rowkey查询  
            scan.addColumn("docsave".getBytes(), "docpath".getBytes()); 
            Result r = table.get(scan);  
            if(r.isEmpty())
            {
            	System.out.println("return NULL get by " + docid +" from "+InfoTableName);
            	return "";
            }
            else
            {
	            System.out.println("GetDirByDocid docid=" + new String(r.getRow()));  
	            docpath = new String(r.list().get(0).getValue());
	            System.out.println("return docpath:" + docpath);            
	            return docpath;            	
            }

        } catch (IOException e) {  
            e.printStackTrace();  
        }
		return docpath; 
    	
    }
    
    
    //对外接口
    public static int GetDocinfoListByKeyword(String keyword, StringBuffer outputTmp)
    {
    	outputTmp.delete(0, outputTmp.length());
        String[] docidlist= GetDocidlistByKeyword(keyword);
        if(null == docidlist)
        {
        	System.out.println("Cannot Get record by keyword:"+keyword);
        	return -1;
        }
        
        HTablePool pool = new HTablePool(configuration, 1000);  
        HTable table = (HTable) pool.getTable(InfoTableName);  
        for (String docid:docidlist)
        {
	        String docinfo= GetDocinfoByKeyword(table, docid);
	        if("" == docinfo)
	        {
	        	System.out.println("Cannot Get record by docid:"+docid);
	        }
	        else
	        {
	        	docinfo+="#";
		        outputTmp.append(docinfo);	
	        }
	                	
        }
        
        return 0;
    	
    }
    
    public static void DeleteHdfsFile(String destpath)
    {

    	
    	try{
	    	FileSystem hdfs=FileSystem.get(configuration);
	
	    	Path dir=new Path(destpath);
	    	boolean isDeleted=hdfs.delete(dir,true);
    	}
    	catch(IOException e)
    	{
    		e.printStackTrace();
    	}
    }
    public static void UploadFile(String destpath, String srcfile)
    {

    	boolean ret = false;
    	try{
	    	FileSystem hdfs=FileSystem.get(configuration);
	
	    	System.out.println("srcpath:"+srcfile+"  destpath:"+destpath);
	        //本地文件
		    	
	    	Path dst=new Path(destpath);
	    	if(!hdfs.exists(dst))
	    	{
	    		ret = hdfs.mkdirs(dst);
	    		System.out.println("mkdir "+dst+" return "+ret);
	    	}

	        Path src =new Path(srcfile);
	
	        System.out.println("Success upload file");
	       
	
	        hdfs.copyFromLocalFile(src, dst);
	

    	}
    	catch(IOException e)
    	{  
            //e.printStackTrace();
    		Log.op("UploadFile fail!  from src:"+srcfile+" to "+destpath);
    	}
    }

  
    /**   未完成  zhu.jinhua 20130320
     * Reads a previously created configuration.
     *
     * @throws ConfigurationException When the XML based configuration is broken.
     */
    public static void LoadPresetVideoData(String PresetDataFileName) throws ConfigurationException {
      
        System.out.println("schema filename: " + PresetDataFileName);
      


      presetDataCfg = new XMLConfiguration(PresetDataFileName);


      PresetdataBaseKey = "records.";
      presetdataList = new ArrayList<TableSchema>();
      readPresetDataFile();
      
      //  System.out.println("preset data read from config: \n  " + presetdataList);
      
    }
    
    
    /**
     * Reads the table schemas from the configuration.
     */
    private static void readPresetDataFile() 
    {
      final int maxTables = presetDataCfg.getMaxIndex(PresetdataBaseKey + "record");
      HTablePool pool = new HTablePool(configuration, 1000);  
      HTable table = (HTable) pool.getTable(InfoTableName);
	  HTable indextable = (HTable) pool.getTable(IndexTableName);
		
		
		
      // iterate over tables
      for (int t = 0; t <= maxTables; t++) 
      {
        final String base = PresetdataBaseKey + "record(" + t + ").";
        final TableSchema ts = new TableSchema();

  	  
        String sdocid		= presetDataCfg.getString(base + "docid");
        String stitle		= presetDataCfg.getString(base + "title");
        String sfilename	= presetDataCfg.getString(base + "filename");
        String sdestpath	= presetDataCfg.getString(base + "destpath");
        String salias		= presetDataCfg.getString(base + "alias");
        String syear		= presetDataCfg.getString(base + "year");
        String sarea		= presetDataCfg.getString(base + "area");
        String svideotype	= presetDataCfg.getString(base + "type");
        String sdirector	= presetDataCfg.getString(base + "director");
        String  sactors 	= presetDataCfg.getString(base + "actors");
        String  skeywords 	= presetDataCfg.getString(base + "keywords");
        String  sintro 		= presetDataCfg.getString(base + "intro");
        String  soutline 	= presetDataCfg.getString(base + "outline");
        String  srcfile 	= presetDataCfg.getString(base + "srcfile");
		String  picfile 	= presetDataCfg.getString(base + "picfile");
        
        
        if(null == sdocid || null == stitle|| null == sfilename || null == sdestpath)
        {
        	System.out.println("docid/title/filename/destpath of record "+t+" cannot be null! next");
        	continue;
        }
        
        String recordinf="get from presetdata title:["+stitle+"] docid:["+sdocid+"] filename:["+sfilename
        		+"] destpath:["+sdestpath+"] alias:["+salias+"] year:["+syear+"] ...";
        System.out.println(recordinf);		
		
        
        
		String docfilename 	= null!=srcfile?srcfile.substring(srcfile.lastIndexOf("/") + 1):null;
		String posterfilename = null!=picfile?picfile.substring(picfile.lastIndexOf("/") + 1):null;
		//String sdestdocpath = (null!=destpath&&null!=docfilename)?destpath +"/"+ docfilename:null;
		//String sdestdocpath = (null!=destpath&&null!=docfilename)?destpath.trim() +"/"+ sdocid.trim():null;
		String sdestdocpath = (null!=sdestpath&&null!=docfilename)?sdestpath.trim():null;
		//String sdestposterpath = (null!=destpath&&null!=docfilename)?destpath +"/"+ posterfilename:null;
		
		System.out.println("sdestdocpath:"+sdestdocpath);
		byte [] docid = Bytes.toBytes(sdocid.trim());
		byte [] title = Bytes.toBytes(stitle.trim());
		byte [] filename = Bytes.toBytes(sfilename.trim());
		byte [] docpath 	= null!=sdestdocpath?Bytes.toBytes(sdestdocpath):null;
		
		
		byte [] alias 		= null!=salias?Bytes.toBytes(salias.trim()):null;
		byte [] year 		= null!=syear?Bytes.toBytes(syear.trim()):null;
		byte [] area 		= null!=sarea?Bytes.toBytes(sarea.trim()):null;
		byte [] videotype 	= null!=svideotype?Bytes.toBytes(svideotype):null;
		byte [] director 	= null!=sdirector?Bytes.toBytes(sdirector.trim()):null;
		byte [] actors 		= null!=sactors?Bytes.toBytes(sactors.trim()):null;
		byte [] keywords 	= null!=skeywords?Bytes.toBytes(skeywords.trim()):null;
		byte [] intro 		= null!=sintro?Bytes.toBytes(sintro.trim()):null;
		byte [] outline 	= null!=soutline?Bytes.toBytes(soutline):null;

		
		
		//byte [] posterpath 	= null!=sdestposterpath?Bytes.toBytes(sdestposterpath.trim()):null;

	
		
		// Create Put
		Put put = new Put(docid);
		put.add(InfoTable_ColName_docinfo.getBytes(), "title".getBytes(), title);
		put.add(InfoTable_ColName_docinfo.getBytes(), "filename".getBytes(), filename);
		put.add(InfoTable_ColName_docsave.getBytes(), "docpath".getBytes(), docpath);
		
		if(null != alias)
		{
			put.add(InfoTable_ColName_docinfo.getBytes(), "alias".getBytes(), alias);
		}
		if(null != year)
		{
			put.add(InfoTable_ColName_docinfo.getBytes(), "year".getBytes(), year);
		}
		if(null != area)
		{
			put.add(InfoTable_ColName_docinfo.getBytes(), "area".getBytes(), area);
		}
		if(null != videotype)
		{
			put.add(InfoTable_ColName_docinfo.getBytes(), "videotype".getBytes(), videotype);
		}
		if(null != director)
		{
			put.add(InfoTable_ColName_docinfo.getBytes(), "director".getBytes(), director);
		}
		if(null != actors)
		{
			put.add(InfoTable_ColName_docinfo.getBytes(), "actor".getBytes(), actors);
		}
		if(null != keywords)
		{
			put.add(InfoTable_ColName_docinfo.getBytes(), "keyword".getBytes(), keywords);
		}
		if(null != intro)
		{
			put.add(InfoTable_ColName_docinfo.getBytes(), "intro".getBytes(), intro);
		}
		if(null != outline)
		{
			put.add(InfoTable_ColName_docinfo.getBytes(), "outline".getBytes(), outline);
		}
		
		/*
		if(null != posterpath)
		{
			put.add(InfoTable_ColName_docsave.getBytes(), "posterpath".getBytes(), posterpath);
		}
		*/


		if(null != sdestdocpath && null!=srcfile)
		{
			// 路径存在则上传文件 , 允许不存在		
			UploadFile(sdestdocpath, srcfile);
		}
		
		
		try {  
			table.put(put);  
		} catch (IOException e) 
		{  
			e.printStackTrace();  
		} 
		
		String [] actorlist= null!=sactors?sactors.toString().split(",|;"):null;
		String [] keywordlist = null!=skeywords?skeywords.toString().split(",|;"):null;
		
		
		
		//Create index
		//create index by filename
		createIndex(indextable, sfilename, sdocid);
		
		//create index by actor
		for(String word :actorlist)
		{
			String sword=word.trim();
			if(sword=="")
			{
				continue;
			}
			createIndex(indextable, sword, sdocid);
		}
		for(String word :keywordlist)
		{
			String sword=word.trim();
			if(sword=="")
			{
				continue;
			}
			createIndex(indextable, sword, sdocid);
		}
		
				
      }
    }

       
    private static void createIndex(HTable indextable, String keyword, String docid_new) 
    {
		System.out.println("words :" + keyword+"  docid:"+docid_new);
		

		try{
		Get g = new Get(Bytes.toBytes(keyword));  
        g.addColumn(Bytes.toBytes(IndexTable_ColName_docidset), Bytes.toBytes("docidset"));  
        Result r=indextable.get(g);  
        
        String docids_old="";
        if(!r.isEmpty())
        {
        	//检查原来的数据
        	docids_old = new String(r.list().get(0).getValue());
        	String [] docidlist_old = docids_old.split(";");
        	for(String docid_old:docidlist_old)
        	{
        		if (docid_old.equals(docid_new))
        		{
        			//原来已经存在了
        			return;
        		}
        	}
        	
        	docids_old+=";";
        }
		
        byte[] docids_save = Bytes.toBytes(docids_old+docid_new);
		Put put_index = new Put(Bytes.toBytes(keyword));
		put_index.add(IndexTable_ColName_docidset.getBytes(), "docidset".getBytes(), docids_save);
		indextable.put(put_index);}
		catch(IOException e) 
		{  
			e.printStackTrace();  
		}
    }
  
    /** 
     * 删除一张表 
     * @param tableName 
     */  
    public static void dropTable(String tablename) {  
        try {  
            HBaseAdmin admin = new HBaseAdmin(configuration);  
            admin.disableTable(tablename);  
            admin.deleteTable(tablename);  
        } catch (MasterNotRunningException e) {  
            e.printStackTrace();  
        } catch (ZooKeeperConnectionException e) {  
            e.printStackTrace();  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
  
    }  
    /** 
     * 根据 rowkey删除一条记录 
     * @param tablename 
     * @param rowkey 
     */  
     public static void deleteRow(String tablename, String rowkey)  {  
        try {  
            HTable table = new HTable(configuration, tablename);  
            List list = new ArrayList();  
            Delete d1 = new Delete(rowkey.getBytes());  
            list.add(d1);  
              
            table.delete(list);  
            System.out.println("删除行成功!");  
              
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
          
  
    }  
  
     /** 
      * 组合条件删除 
      * @param tablename 
      * @param rowkey 
      */  
     public static void deleteByCondition(String tablename, String rowkey)  {  
            //目前还没有发现有效的API能够实现 根据非rowkey的条件删除 这个功能能，还有清空表全部数据的API操作  
  
    }  
  
  
    /** 
     * 查询所有数据 
     * @param tableName 
     */  
    public static void QueryAll(String tableName) {  
        HTablePool pool = new HTablePool(configuration, 1000);  
        HTable table = (HTable) pool.getTable(tableName);  
        try {  
            ResultScanner rs = table.getScanner(new Scan());  
            for (Result r : rs) {  
                System.out.println("获得到rowkey:" + new String(r.getRow()));  
                for (KeyValue keyValue : r.raw()) {  
                    System.out.println(new String(keyValue.getFamily()) + ":"  
                            + new String(keyValue.getQualifier()) + " timestamp " + keyValue.getTimestamp() 
                            + "\t:" + new String(keyValue.getValue()));  
                }  
            }  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
    }  
  
    /** 
     * 单条件查询,根据rowkey查询唯一一条记录 
     * @param tableName 
     */  
    public static String GetDocinfoByKeyword(HTable table, String key) {  
    	String strret="";

        try {  
            Get scan = new Get(key.getBytes());// 根据rowkey查询  
            Result r = table.get(scan);  
            if(r.isEmpty())
            {
            	System.out.println("GetDocinfoByKeyword:return NULL get by " + key);
            	return "";
            }
            else
            {
	            strret="docid:"+key+"; ";
	            for (KeyValue keyValue : r.raw()) 
	            {
	            	strret+=(new String(keyValue.getQualifier()))+":";
	            	strret+=new String(keyValue.getValue())+";";
	            	
	            }  
	            System.out.println("GetDocinfoByKeyword:query by:" + new String(r.getRow())+" return :\"" + strret+"\"");            
	            return strret;            	
            }

        } catch (IOException e) {  
            e.printStackTrace();  
        }
		return strret;  
    }  
  
    /** 
     * 单条件按查询，查询多条记录 
     * @param tableName 
     */  
    public static void QueryByCondition2(String tableName, String ColFamily, String ColQua, String key) {  
  
        try {  
            HTablePool pool = new HTablePool(configuration, 1000);  
            HTable table = (HTable) pool.getTable(tableName);  
            Filter filter = new SingleColumnValueFilter(Bytes  
                    .toBytes(ColFamily), null, CompareOp.EQUAL, Bytes  
                    .toBytes(key)); // 当列column1的值为aaa时进行查询  
            Scan s = new Scan();  
            s.setFilter(filter);  
            ResultScanner rs = table.getScanner(s);  
            for (Result r : rs) {  
                System.out.println("获得到rowkey:" + new String(r.getRow()));  
                for (KeyValue keyValue : r.raw()) {  
                    System.out.println("列：" + new String(keyValue.getFamily())  
                            + "====值:" + new String(keyValue.getValue()));  
                }  
            }  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
  
    }  
  
    /** 
     * 组合条件查询 
     * @param tableName 
     */  
    public static void QueryByCondition3(String tableName) {  
  
        try {  
            HTablePool pool = new HTablePool(configuration, 1000);  
            HTable table = (HTable) pool.getTable(tableName);  
  
            List<Filter> filters = new ArrayList<Filter>();  
  
            Filter filter1 = new SingleColumnValueFilter(Bytes  
                    .toBytes("column1"), null, CompareOp.EQUAL, Bytes  
                    .toBytes("aaa"));  
            filters.add(filter1);  
  
            Filter filter2 = new SingleColumnValueFilter(Bytes  
                    .toBytes("column2"), null, CompareOp.EQUAL, Bytes  
                    .toBytes("bbb"));  
            filters.add(filter2);  
  
            Filter filter3 = new SingleColumnValueFilter(Bytes  
                    .toBytes("column3"), null, CompareOp.EQUAL, Bytes  
                    .toBytes("ccc"));  
            filters.add(filter3);  
  
            FilterList filterList1 = new FilterList(filters);  
  
            Scan scan = new Scan();  
            scan.setFilter(filterList1);  
            ResultScanner rs = table.getScanner(scan);  
            for (Result r : rs) {  
                System.out.println("获得到rowkey:" + new String(r.getRow()));  
                for (KeyValue keyValue : r.raw()) {  
                    System.out.println("列：" + new String(keyValue.getFamily())  
                            + "====值:" + new String(keyValue.getValue()));  
                }  
            }  
            rs.close();  
  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
  
    }  
  
} 
